package storm.bolt;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class OrderSplitBolt extends BaseBasicBolt{
    public void prepare(Map conf, TopologyContext context){

    }
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        //将每一个tuple转为字节码
        byte[]bytes=tuple.getBinary(0);
        //byte->String
        String line = new String(bytes);
        //将每个元组以”，“分割
        String[] str = line.split("\",\"");
        //获取每条元组所需的信息
        String order_id =str[0];
        String book_id=str[1];
        String book_name =str[2];
        String book_type =str[3];
        String book_num=str[4];
        String book_price=str[5];
        String  book_press=str[6];
        String id=str[7];

        //System.out.println(line);
        //将信息重组成元组的value传至下一个bolt中进行输出和贮存
        collector.emit(new Values(order_id,book_id,book_name,book_type,book_num,book_price,book_press,id));
    }
    //向下一个bolt传输元组的key的字段名称与emit配合具体看ppt24、25
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("order_id", "book_id", "book_name","book_type","book_num","book_price",
                "book_press","id"));
    }
}
